#include "in.h"

struct MQDelayMessage *FindRecentSendTimestampDelayMessage( struct MQStore *store )
{
	struct rb_node		*first_tree_node = NULL ;
	
	first_tree_node = rb_first( & (store->delay_msg_send_timestamp_tree) ) ;
	if( first_tree_node == NULL )
		return NULL;
	
	return container_of( first_tree_node , struct MQDelayMessage , delay_msg_send_timestamp_tree_node ) ;
}

struct MQDelayMessage *CreateDelayMessage( struct TcpdaemonServerEnvironment *p_env , struct AppendBuffer *comm_data , struct MQFileMapping *last_delay_msg_file , char *fill_ptr , uint64_t delay_msg_offset , uint32_t delay_seconds )
{
	struct MQDelayMessage	*delay_msg = NULL ;
	struct itimerspec	its ;
	
	int			nret = 0 ;
	
	delay_msg = (struct MQDelayMessage *)malloc( sizeof(struct MQDelayMessage) ) ;
	if( delay_msg == NULL )
	{
		ERRORLOG( "malloc size[%zu] failed , errno[%d]" , sizeof(struct MQDelayMessage) , errno )
		MEMPACK_RESULT( return NULL , MQSDK_ERROR_MALLOC , comm_data )
		return NULL;
	}
	else
	{
		DEBUGLOG( "malloc ok , size[%zu]" , sizeof(struct MQDelayMessage) )
	}
	memset( delay_msg , 0x00 , sizeof(struct MQDelayMessage) );
	
	delay_msg->custom_evnet_session = CUSTOMEVENT1_FD_SESSION ;
	delay_msg->delay_msg_offset = delay_msg_offset ;
	delay_msg->delay_seconds = delay_seconds ;
	
	delay_msg->delay_timer = timerfd_create( CLOCK_MONOTONIC , O_NONBLOCK ) ;
	if( delay_msg->delay_timer == -1 )
	{
		FATALLOG( "timerfd_create failed , errno[%d]" , errno )
		MEMPACK_RESULT( return NULL , MQSDK_ERROR_TIMERFD_CREATE , comm_data )
		free( delay_msg );
		return NULL;
	}
	else
	{
		DEBUGLOG( "timerfd_create ok , delay_timer[%d]" , delay_msg->delay_timer )
	}
	
	its.it_value.tv_sec = delay_msg->delay_seconds ;
	its.it_value.tv_nsec = 0 ;
	its.it_interval.tv_sec = 0 ;
	its.it_interval.tv_nsec = 0 ;
	nret = timerfd_settime( delay_msg->delay_timer , 0 , & its , NULL ) ;
	if( nret == -1 )
	{
		FATALLOG( "timerfd_settime failed , errno[%d] , delay_timer[%d]" , errno , delay_msg->delay_timer );
		MEMPACK_RESULT( return NULL , MQSDK_ERROR_TIMERFD_SETTIME , comm_data )
		free( delay_msg );
		return NULL;
	}
	else
	{
		INFOLOG( "timerfd_settime ok , delay_timer[%d] - it_value[%ld][%ld] it_interval[%ld][%ld]" , delay_msg->delay_timer , its.it_value.tv_sec,its.it_value.tv_nsec , its.it_interval.tv_sec,its.it_interval.tv_nsec );
	}
	
	if( p_env )
	{
		DEBUGLOG( "TDAddCustemSessionToEventBus delay_msg->delay_timer[%d]" , delay_msg->delay_timer )
		TDAddCustemSessionToEventBus( p_env , delay_msg->delay_timer , delay_msg );
	}
	
	delay_msg->delay_msg_file = last_delay_msg_file ;
	delay_msg->delay_msg_file_unit_base = fill_ptr ;
	
	return delay_msg;
}

void DestroyDelayMessage( struct TcpdaemonServerEnvironment *p_env , struct MQDelayMessage *delay_msg )
{
	if( p_env )
	{
		DEBUGLOG( "TDRemoveCustemSessionFromEventBus delay_msg->delay_timer[%d]" , delay_msg->delay_timer )
		TDRemoveCustemSessionFromEventBus( p_env , delay_msg->delay_timer );
	}
	
	DEBUGLOG( "close delay_msg->delay_timer[%d]" , delay_msg->delay_timer )
	close( delay_msg->delay_timer );
	
	DEBUGLOG( "free delay_msg[%p]" , delay_msg )
	free( delay_msg );
	
	return;
}

void MemSet_DelayMessageDataUnit_is_sent( char *unit_offset_ptr , char is_sent )
{
	char		*p_is_sent = unit_offset_ptr ;
	
	p_is_sent += sizeof(uint64_t) + sizeof(uint32_t) ;
	(*p_is_sent) = is_sent ;
	
	return;
}

size_t CalcDelayMessageDataUnitSize( uint8_t topic_len , uint16_t tags_len , uint32_t msg_data_len )
{
	return sizeof(uint64_t) + sizeof(uint32_t) + 1 + sizeof(uint8_t) + topic_len + sizeof(uint16_t) + tags_len + sizeof(uint32_t) + msg_data_len ;
}

uint64_t MemWrite_DelayMessageDataUnit( char *unit_offset_ptr , uint64_t delay_msg_id , uint32_t delay_seconds , char is_sent , uint8_t store_name_len , char *store_name , uint8_t topic_name_len , char *topic_name , uint16_t tags_len , char *tags , uint32_t msg_data_len , char *msg_data )
{
	char		*p = unit_offset_ptr ;
	uint64_t	pack_len ;
	
	*(uint64_t*)p = hton64(delay_msg_id) ;
	if( is_sent == MQ_DELAY_MSG_ROLLBACK )
		return 0;
	p += sizeof(uint64_t) ;
	
	*(uint32_t*)p = hton32(delay_seconds) ;
	p += sizeof(uint32_t) ;
	
	*(char*)p = is_sent ;
	p++;
	
	*(uint8_t*)p = store_name_len ;
	p += sizeof(uint8_t) ;
	
	memcpy( p , store_name , store_name_len );
	p += store_name_len ;
	
	*(uint8_t*)p = topic_name_len ;
	p += sizeof(uint8_t) ;
	
	memcpy( p , topic_name , topic_name_len );
	p += topic_name_len ;
	
	*(uint16_t*)p = hton16(tags_len) ;
	p += sizeof(uint16_t) ;
	
	memcpy( p , tags , tags_len );
	p += tags_len ;
	
	*(uint32_t*)p = hton32(msg_data_len) ;
	p += sizeof(uint32_t) ;
	
	memcpy( p , msg_data , msg_data_len );
	p += msg_data_len ;
	
	*(uint64_t*)p = hton64(0) ;
	
	pack_len = p - unit_offset_ptr ;
	
	return pack_len;
}

uint64_t MemRead_DelayMessageDataUnit( char *unit_offset_ptr , uint64_t *p_delay_msg_id , uint32_t *p_delay_seconds , char *p_is_sent , uint8_t *p_store_name_len , char **pp_store_name , uint8_t *p_topic_name_len , char **pp_topic_name , uint16_t *p_tags_len , char **pp_tags , uint32_t *p_msg_data_len , char **pp_msg_data )
{
	char		*p = unit_offset_ptr ;
	uint64_t	delay_msg_id ;
	uint32_t	delay_seconds ;
	char		is_sent ;
	uint8_t		store_name_len ;
	uint8_t		topic_name_len ;
	uint16_t	tags_len ;
	uint32_t	msg_data_len ;
	uint64_t	unpack_len ;
	
	delay_msg_id = ntoh64(*(uint64_t*)p) ;
	if( p_delay_msg_id )
		(*p_delay_msg_id) = delay_msg_id ;
	if( delay_msg_id == 0 || delay_msg_id == UINT64_MAX )
		return delay_msg_id;
	if( p_delay_msg_id )
		(*p_delay_msg_id) = delay_msg_id ;
	p += sizeof(uint64_t) ;
	
	delay_seconds = ntoh32(*(uint32_t*)p) ;
	if( p_delay_seconds )
		(*p_delay_seconds) = delay_seconds ;
	p += sizeof(uint32_t) ;
	
	is_sent = *(char*)p ;
	if( p_is_sent )
		(*p_is_sent) = is_sent ;
	p++;
	
	store_name_len = *(uint8_t*)p ;
	if( p_store_name_len )
		(*p_store_name_len) = store_name_len ;
	p += sizeof(uint8_t) ;
	
	if( pp_store_name )
		(*pp_store_name) = p ;
	p += store_name_len ;
	
	topic_name_len = *(uint8_t*)p ;
	if( p_topic_name_len )
		(*p_topic_name_len) = topic_name_len ;
	p += sizeof(uint8_t) ;
	
	if( pp_topic_name )
		(*pp_topic_name) = p ;
	p += topic_name_len ;
	
	tags_len = ntoh16(*(uint16_t*)p) ;
	if( p_tags_len )
		(*p_tags_len) = tags_len ;
	p += sizeof(uint16_t) ;
	
	if( pp_tags )
		(*pp_tags) = p ;
	p += tags_len ;
	
	msg_data_len = ntoh32(*(uint32_t*)p) ;
	if( p_msg_data_len )
		(*p_msg_data_len) = msg_data_len ;
	p += sizeof(uint32_t) ;
	
	if( pp_msg_data )
		(*pp_msg_data) = p ;
	p += msg_data_len ;
	
	unpack_len = p - unit_offset_ptr ;
	
	return unpack_len;
}

int CreateDelayMessageDataFile( struct AppendBuffer *comm_data , struct MQBroker *broker , struct MQStore *store , size_t begin_offset , struct MQFileMapping **pp_delay_msg_file , size_t *p_file_size )
{
	struct MQFileMapping	*delay_msg_file = NULL ;
	int			file_fd ;
	off_t			seek_file_offset ;
	ssize_t			write_file_len ;
	
	int			nret = 0 ;
	
	delay_msg_file = (struct MQFileMapping *)malloc( sizeof(struct MQFileMapping) ) ;
	if( delay_msg_file == NULL )
	{
		ERRORLOG( "malloc failed , errno[%d]" , errno )
		if( comm_data )
			MEMPACK_RESULT( return -11 , MQSDK_ERROR_MALLOC , comm_data )
		return -1;
	}
	memset( delay_msg_file , 0x00 , sizeof(struct MQFileMapping) );
	
	memset( delay_msg_file->path_file_name , 0x00 , sizeof(delay_msg_file->path_file_name) );
	snprintf( delay_msg_file->path_file_name , sizeof(delay_msg_file->path_file_name)-1 , "%s/%.*s/%020zu.cmqd" , broker->data_path , store->store_name_len,store->store_name , begin_offset );
	file_fd = open( delay_msg_file->path_file_name , O_CREAT|O_EXCL|O_RDWR|O_TRUNC , 00777 ) ;
	if( file_fd == -1 )
	{
		free( delay_msg_file );
		if( comm_data )
			MEMPACK_RESULT( return -12 , MQSDK_ERROR_CREATE_DELAY_MSG_FILE , comm_data )
		return -2;
	}
	else
	{
		DEBUGLOG( "open[%s] ok" , delay_msg_file->path_file_name )
	}
	
	delay_msg_file->begin_offset = begin_offset ;
	delay_msg_file->file_size = broker->delay_msg_file_size ;
	
	seek_file_offset = lseek( file_fd , delay_msg_file->file_size-1 , SEEK_SET ) ;
	if( seek_file_offset == -1 )
	{
		ERRORLOG( "lseek[%s] offset[%zu] failed , errno[%d]" , delay_msg_file->path_file_name , delay_msg_file->file_size-1 , errno )
		close( file_fd );
		unlink( delay_msg_file->path_file_name );
		free( delay_msg_file );
		if( comm_data )
			MEMPACK_RESULT( return -13 , MQSDK_ERROR_LSEEK_DELAY_MSG_FILE , comm_data )
		return -3;
	}
	else
	{
		DEBUGLOG( "lseek[%s] offset[%zu] ok" , delay_msg_file->path_file_name , delay_msg_file->file_size-1 )
	}
	
	write_file_len = write( file_fd , "\0" , 1 ) ;
	if( write_file_len == -1 )
	{
		ERRORLOG( "write[%s] failed , errno[%d]" , delay_msg_file->path_file_name , errno )
		close( file_fd );
		unlink( delay_msg_file->path_file_name );
		free( delay_msg_file );
		if( comm_data )
			MEMPACK_RESULT( return -14 , MQSDK_ERROR_LWRITE_DELAY_MSG_FILE , comm_data )
		return -4;
	}
	else
	{
		DEBUGLOG( "write[%s] ok" , delay_msg_file->path_file_name )
	}
	
	delay_msg_file->mmap_size = delay_msg_file->file_size ;
	delay_msg_file->file_mmap = mmap( NULL , delay_msg_file->file_size , PROT_READ|PROT_WRITE , MAP_SHARED , file_fd , 0 ) ;
	if( delay_msg_file->file_mmap == (void*)(-1) )
	{
		ERRORLOG( "mmap[%s] failed , errno[%d]" , delay_msg_file->path_file_name , errno )
		close( file_fd );
		unlink( delay_msg_file->path_file_name );
		free( delay_msg_file );
		if( comm_data )
			MEMPACK_RESULT( return -15 , MQSDK_ERROR_LWRITE_DELAY_MSG_FILE , comm_data )
		return -5;
	}
	else
	{
		DEBUGLOG( "mmap[%s] size[%zu] ok , addr[0x%X]" , delay_msg_file->path_file_name , delay_msg_file->file_size , delay_msg_file->file_mmap )
	}
	
	close( file_fd );
	
	memset( delay_msg_file->file_mmap , 0x00 , delay_msg_file->file_size );
	delay_msg_file->fill_ptr = delay_msg_file->file_mmap ;
	delay_msg_file->over_end_ptr = delay_msg_file->file_mmap + delay_msg_file->file_size ;
	DEBUGLOG( "delay_msg_file->over_end_ptr[0x%X]" , delay_msg_file->over_end_ptr )
	
	DEBUGLOG( "write delay_msg_file head" )
	memcpy( delay_msg_file->fill_ptr+MQ_DELAY_MSG_FILE_HEADER_FORMAT_MAGIC_OFFSET , MQ_DELAY_MSG_FILE_HEADER_FORMAT_MAGIC_VALUE , MQ_DELAY_MSG_FILE_HEADER_FORMAT_MAGIC_LENGTH );
	*(delay_msg_file->fill_ptr+MQ_DELAY_MSG_FILE_HEADER_FORMAT_VERSION_OFFSET) = MQ_DELAY_MSG_FILE_RESTORE_FORMAT_VERSION_1 ;
	*((uint64_t*)(delay_msg_file->fill_ptr+MQ_DELAY_MSG_FILE_HEADER_FORMAT_MSG_OFFSET_OFFSET)) = hton64(begin_offset) ;
	*((uint64_t*)(delay_msg_file->fill_ptr+MQ_DELAY_MSG_FILE_HEADER_FORMAT_ORI_FILE_SIZE_OFFSET)) = hton64(delay_msg_file->file_size) ;
	*(delay_msg_file->fill_ptr+MQ_DELAY_MSG_FILE_HEADER_FORMAT_IS_CLEAN_OFFSET) = MQ_DELAY_MSG_FILE_HEADER_FORMAT_NOT_CLEAN ;
	delay_msg_file->fill_ptr += MQ_DELAY_MSG_FILE_HEAD_SIZE ;
	DEBUGLOG( "delay_msg_file->fill_ptr[0x%X]" , delay_msg_file->fill_ptr )
	
	store->max_delay_msg_offset += MQ_DELAY_MSG_FILE_HEAD_SIZE ;
	DEBUGLOG( "store->max_delay_msg_offset[%zu]" , store->max_delay_msg_offset )
	
	DEBUGLOG( "add delay_msg_file[%s] to store's msg_file_list" , delay_msg_file->path_file_name )
	list_add_tail( & (delay_msg_file->file_list_node) , & (store->delay_msg_file_list) );
	store->delay_msg_file_list_nodes_count++;
	if( store->first_not_sent_delay_msg_file == NULL )
	{
		DEBUGLOG( "set store->first_not_sent_delay_msg_file[%p]" , store->first_not_sent_delay_msg_file )
		store->first_not_sent_delay_msg_file = delay_msg_file ;
	}
	DEBUGLOG( "set store->last_not_sent_delay_msg_file[%p] to [%p]" , store->last_not_sent_delay_msg_file , delay_msg_file )
	store->last_not_sent_delay_msg_file = delay_msg_file ;
	
	if( pp_delay_msg_file )
		(*pp_delay_msg_file) = delay_msg_file ;
	if( p_file_size )
		(*p_file_size) = delay_msg_file->file_size ;
	return 0;
}

int LoadDelayMessageDataFile( struct AppendBuffer *comm_data , struct MQBroker *broker , struct MQStore *store , size_t begin_offset , char *filename , struct MQFileMapping **pp_delay_msg_file , size_t *p_file_size )
{
	struct MQFileMapping	*delay_msg_file = NULL ;
	int			file_fd ;
	
	int			nret = 0 ;
	
	delay_msg_file = (struct MQFileMapping *)malloc( sizeof(struct MQFileMapping) ) ;
	if( delay_msg_file == NULL )
	{
		ERRORLOG( "malloc failed , errno[%d]" , errno )
		if( comm_data )
			MEMPACK_RESULT( return -11 , MQSDK_ERROR_MALLOC , comm_data )
		return -1;
	}
	memset( delay_msg_file , 0x00 , sizeof(struct MQFileMapping) );
	
	memset( delay_msg_file->path_file_name , 0x00 , sizeof(delay_msg_file->path_file_name) );
	if( filename == NULL )
		snprintf( delay_msg_file->path_file_name , sizeof(delay_msg_file->path_file_name)-1 , "%s/%.*s/%020zu.cmqd" , broker->data_path , store->store_name_len,store->store_name , begin_offset );
	else
		strncpy( delay_msg_file->path_file_name , filename , sizeof(delay_msg_file->path_file_name)-1 );
	DEBUGLOG( "try to open[%s]" , delay_msg_file->path_file_name )
	file_fd = open( delay_msg_file->path_file_name , O_RDWR , 00777 ) ;
	if( file_fd == -1 )
	{
		DEBUGLOG( "open[%s] not exist" , delay_msg_file->path_file_name )
		free( delay_msg_file );
		if( comm_data )
			MEMPACK_RESULT( return -12 , MQSDK_ERROR_CREATE_DELAY_MSG_FILE , comm_data )
		return -2;
	}
	else
	{
		DEBUGLOG( "open[%s] ok" , delay_msg_file->path_file_name )
	}
	
	delay_msg_file->mmap_size = GetFileSize( NULL , file_fd ) ;
	if( delay_msg_file->mmap_size < MQ_DELAY_MSG_FILE_HEAD_SIZE )
	{
		ERRORLOG( "file size[%z] invalid" , delay_msg_file->mmap_size )
		close( file_fd );
		unlink( delay_msg_file->path_file_name );
		free( delay_msg_file );
		if( comm_data )
			MEMPACK_RESULT( return -13 , MQSDK_ERROR_DELAY_MSG_FILE_SIZE_INVALID , comm_data )
		return -3;
	}
	delay_msg_file->file_mmap = mmap( NULL , delay_msg_file->mmap_size , PROT_READ|PROT_WRITE , MAP_SHARED , file_fd , 0 ) ;
	if( delay_msg_file->file_mmap == (void*)(-1) )
	{
		ERRORLOG( "mmap[%s] failed , errno[%d]" , delay_msg_file->path_file_name , errno )
		close( file_fd );
		unlink( delay_msg_file->path_file_name );
		free( delay_msg_file );
		if( comm_data )
			MEMPACK_RESULT( return -13 , MQSDK_ERROR_LWRITE_DELAY_MSG_FILE , comm_data )
		return -3;
	}
	else
	{
		if( filename == NULL )
			delay_msg_file->begin_offset = begin_offset ;
		else
			delay_msg_file->begin_offset = *((uint64_t*)(delay_msg_file->file_mmap+5)) ;
		delay_msg_file->file_size = ntoh64(*((uint64_t*)(delay_msg_file->file_mmap+13))) ;
		DEBUGLOG( "mmap[%s] mmap_size[%zu] file_size[%zu] ok , addr[0x%X]" , delay_msg_file->path_file_name , delay_msg_file->mmap_size , delay_msg_file->file_size , delay_msg_file->file_mmap )
	}
	
	if( p_file_size )
	{
		(*p_file_size) = delay_msg_file->file_size ;
		DEBUGLOG( "(*p_file_size)[%zu]" , (*p_file_size) )
	}
	
	close( file_fd );
	
	if( *(delay_msg_file->file_mmap+MQ_DELAY_MSG_FILE_HEADER_FORMAT_IS_CLEAN_OFFSET) == MQ_DELAY_MSG_FILE_HEADER_FORMAT_NOT_CLEAN )
	{
		delay_msg_file->over_end_ptr = delay_msg_file->file_mmap + delay_msg_file->mmap_size ;
		delay_msg_file->fill_ptr = delay_msg_file->file_mmap + MQ_DELAY_MSG_FILE_HEAD_SIZE ;
		
		if( store )
		{
			INFOLOG( "add delay_msg_file[%s] to store's delay_msg_file_list" , delay_msg_file->path_file_name )
			list_add_tail( & (delay_msg_file->file_list_node) , & (store->delay_msg_file_list) );
			store->delay_msg_file_list_nodes_count++;
			if( store->first_not_sent_delay_msg_file == NULL )
			{
				DEBUGLOG( "set store->first_not_sent_delay_msg_file[%p]" , delay_msg_file )
				store->first_not_sent_delay_msg_file = delay_msg_file ;
			}
			DEBUGLOG( "set store->last_not_sent_delay_msg_file[%p] to [%p]" , store->last_not_sent_delay_msg_file , delay_msg_file )
			store->last_not_sent_delay_msg_file = delay_msg_file ;
		}
	}
	else
	{
		munmap( delay_msg_file->file_mmap , delay_msg_file->mmap_size );
		free( delay_msg_file ); delay_msg_file = NULL ;
	}
	
	if( pp_delay_msg_file )
	{
		(*pp_delay_msg_file) = delay_msg_file ;
		DEBUGLOG( "ret (*pp_delay_msg_file)[%p]" , (*pp_delay_msg_file) )
	}
	
	return 0;
}

int LoadAllDelayMessageDataFiles( struct MQBroker *broker , struct MQStore *store )
{
	size_t			begin_size ;
	struct MQFileMapping	*delay_msg_file = NULL ;
	size_t			file_size ;
	
	uint64_t		unpack_len ;
	uint32_t		delay_seconds ;
	char			is_sent ;
	struct MQDelayMessage	*delay_msg = NULL ;
	
	int			nret = 0 ;
	
	begin_size = 0 ;
	for( ; ; )
	{
		nret = LoadDelayMessageDataFile( NULL , broker , store , begin_size , NULL , & delay_msg_file , & file_size ) ;
		if( nret == -2 )
		{
			break;
		}
		else if( nret )
		{
			ERRORLOG( "LoadDelayMessageDataFile failed" )
			break;
		}
		else
		{
			INFOLOG( "LoadDelayMessageDataFile ok , begin_size[%zu] file_size[%zu]" , begin_size , file_size )
		}
		
		for( ; ; )
		{
			unpack_len = MemRead_DelayMessageDataUnit( delay_msg_file->fill_ptr , NULL , & delay_seconds , & is_sent , NULL , NULL , NULL , NULL , NULL , NULL , NULL , NULL ) ;
			if( unpack_len == 0 || unpack_len == UINT64_MAX )
			{
				store->max_delay_msg_offset = delay_msg_file->fill_ptr-delay_msg_file->file_mmap + delay_msg_file->begin_offset ;
				break;
			}
			else if( is_sent == MQ_DELAY_MSG_WAIT_TO_SEND )
			{
				delay_msg = CreateDelayMessage( NULL , NULL , delay_msg_file , delay_msg_file->fill_ptr , delay_msg_file->fill_ptr-delay_msg_file->file_mmap + delay_msg_file->begin_offset , delay_seconds ) ;
				if( delay_msg == NULL )
				{
					ERRORLOG( "CreateDelayMessage failed" )
					break;
				}
				else
				{
					DEBUGLOG( "CreateDelayMessage ok" )
				}
				
				LinkDelayMessageOffsetTreeNode( store , delay_msg );
				LinkDelayMessageSendTimestampTreeNode( store , delay_msg );
			}
			
			delay_msg_file->fill_ptr += unpack_len ;
			store->max_delay_msg_offset += unpack_len ;
		}
		
		begin_size += file_size ;
	}
	
	return 0;
}

int UnloadDelayMessageDataFile( struct MQBroker *broker , struct MQStore *store , struct MQFileMapping *delay_msg_file )
{
	munmap( delay_msg_file->file_mmap , delay_msg_file->file_size );
	
	list_del( & (delay_msg_file->file_list_node) );
	store->delay_msg_file_list_nodes_count--;
	
	free( delay_msg_file );
	
	return 0;
}

int CleanDelayMessageDataFile( struct MQBroker *broker , struct MQStore *store , struct MQFileMapping *delay_msg_file )
{
	char		path_file_name[ PATH_MAX ] ;
	
	int		nret = 0 ;
	
	strcpy( path_file_name , delay_msg_file->path_file_name );
	
	DEBUGLOG( "clean msg_file[%s]" , path_file_name )
	*(delay_msg_file->file_mmap+MQ_DELAY_MSG_FILE_HEADER_FORMAT_IS_CLEAN_OFFSET) = MQ_DELAY_MSG_FILE_HEADER_FORMAT_CLEANNED ;
	
	DEBUGLOG( "UnloadDelayMessageDataFile ... store[%s] delay_msg_file[%s]" , store->store_name , path_file_name )
	nret = UnloadDelayMessageDataFile( broker , store , delay_msg_file ) ;
	if( nret )
	{
		ERRORLOG( "UnloadDelayMessageDataFile failed[%d]" , nret )
		return nret;
	}
	else
	{
		DEBUGLOG( "UnloadDelayMessageDataFile ok" )
	}
	
	DEBUGLOG( "truncate msg_file[%s]" , path_file_name )
	truncate( path_file_name , MQ_DELAY_MSG_FILE_HEAD_SIZE );
	
	return 0;
}

int UnloadAllDelayMessageDataFiles( struct MQBroker *broker , struct MQStore *store )
{
	struct MQFileMapping	*delay_msg_file = NULL ;
	struct MQFileMapping	*next_delay_msg_file = NULL ;
	
	list_for_each_entry_safe( delay_msg_file , next_delay_msg_file , & (store->delay_msg_file_list) , struct MQFileMapping , file_list_node )
	{
		UnloadDelayMessageDataFile( broker , store , delay_msg_file );
	}
	
	return 0;
}

